package com.shujia

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import java.util.Properties

object Demo01KafkaProducer {
  def main(args: Array[String]): Unit = {
    // 1、建立连接
    val properties = new Properties()
    // kafka broker列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    // 指定K、V的序列化的类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    val kafkaProducer: KafkaProducer[String, String] = new KafkaProducer[String, String](properties)

    // 2、构建消息格式

    /**
     * 当往一个不存在的Topic中消费或者生产数据时 Kafka会自动创建一个分区数为1，副本数为1的Topic
     */
    val producerRecord = new ProducerRecord[String, String]("test_topic3", "100")

    // 3、发送数据
    kafkaProducer.send(producerRecord)
    kafkaProducer.flush()

    // 4、关闭连接
    kafkaProducer.close()

  }

}
